package com.sheepone.distributor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sheepone.distributor.event.DistributeEvent;
import com.sheepone.distributor.event.EventContext;
import com.sheepone.distributor.thread.DelayEventQueue;
import com.sheepone.distributor.thread.NamedThreadFactoryManager;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;


/**
 * @author Miss.杨
 * @description
 * @since 2023-12-07 21:53
 */
public class DistributorLine {

    private final long sleepNanos = 5;
    private final ExecutorService executorService;
    private final DelayEventQueue<EventTranslator<DistributeEvent>> delayQueue;

    private final Disruptor<DistributeEvent> distributorLine;

    public DistributorLine(String name, int size, DelayEventQueue<EventTranslator<DistributeEvent>> delayQueue) {
        this.delayQueue = delayQueue;
        executorService = Executors.newSingleThreadExecutor(NamedThreadFactoryManager.INSTANCE.acquire(name + "Delay"));
        distributorLine = new Disruptor<>(DistributeEvent::new,
                size,
                r -> {
                    Thread thread = new Thread(r);
                    thread.setName(name);
                    thread.setDaemon(true);
                    return thread;
                },
                ProducerType.MULTI,
                new YieldingWaitStrategy()
        );
        distributorLine.handleEventsWith(WorkerStationManager.INSTANCE);
    }

    public void publish(String targetStation, String targetTopic, EventContext context) {
        EventTranslator<DistributeEvent> translator = (e, s) -> {
            while (true) {
                if (e.isAvailable() && e.doing()) {
                    e.setContext(context);
                    e.setTargetStation(targetStation);
                    e.setTargetTopic(targetTopic);
                    break;
                }
            }
        };
        if (!distributorLine.getRingBuffer().tryPublishEvent(translator)) {
            try {
                delayQueue.put(translator);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public void stop() {
        distributorLine.halt();
        executorService.shutdown();
    }

    public long hasPublishNum() {
        return distributorLine.getCursor() + 1;
    }

    public void start() {
        executorService.submit(() -> {
            while (true) {
                while (!delayQueue.isEmpty()) {
                    try {
                        distributorLine.getRingBuffer().publishEvent(delayQueue.take());
                    } catch (InterruptedException e) {
                        throw e;
                    }
                }
                try {
                    TimeUnit.NANOSECONDS.sleep(sleepNanos);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        distributorLine.start();
    }
}
